package org.mq.mymq.broker.event.listener;

import com.alibaba.fastjson.JSON;
import org.apache.commons.collections4.CollectionUtils;
import org.mq.mymq.broker.cache.CommonCache;
import org.mq.mymq.broker.event.model.ConsumeMsgAckEvent;
import org.mq.mymq.broker.model.MyMqTopicModel;
import org.mq.mymq.broker.rebalance.ConsumerInstance;
import org.mq.mymq.common.coder.TcpMsg;
import org.mq.mymq.common.dto.ConsumeMsgAckReqDTO;
import org.mq.mymq.common.dto.ConsumeMsgAckRespDTO;
import org.mq.mymq.common.enums.AckStatus;
import org.mq.mymq.common.enums.BrokerResponseCode;
import org.mq.mymq.common.event.Listener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;

/**
 * @ClassName ConsumeMsgAckListener
 * @Description 消费端回应ack的监听器
 * @Author jiarong_ye
 * @Date 2025/6/30 17:00
 * @Version 1.0
 */
public class ConsumeMsgAckListener implements Listener<ConsumeMsgAckEvent> {
    private static final Logger log = LoggerFactory.getLogger(ConsumeMsgAckListener.class);

    @Override
    public void onReceive(ConsumeMsgAckEvent event) throws Exception {
        ConsumeMsgAckReqDTO consumeMsgAckReqDTO = event.getConsumeMsgAckReqDTO();
        String topic = consumeMsgAckReqDTO.getTopic();
        String consumerGroup = consumeMsgAckReqDTO.getConsumeGroup();
        Integer queueId = consumeMsgAckReqDTO.getQueueId();
        Integer ackCount = consumeMsgAckReqDTO.getAckCount();
        MyMqTopicModel myMqTopicModel = CommonCache.getMyMqTopicModelsMap().get(topic);
        ConsumeMsgAckRespDTO consumeMsgAckRespDTO = new ConsumeMsgAckRespDTO();
        consumeMsgAckRespDTO.setAckStatus(AckStatus.FAIL.getCode());
        consumeMsgAckRespDTO.setMsgId(event.getMsgId());
        if (myMqTopicModel == null) {
            // topic不存在，ack失败
            event.getChannelHandlerContext().writeAndFlush(new TcpMsg(BrokerResponseCode.BROKER_UPDATE_OFFSET_RESP.getCode(), JSON.toJSONBytes(consumeMsgAckRespDTO)));
            return;
        }
        Map<String, List<ConsumerInstance>> consumeInstanceMap = CommonCache.getConsumerHoldMap().get(topic);
        if (consumeInstanceMap == null || consumeInstanceMap.isEmpty()) {
            event.getChannelHandlerContext().writeAndFlush(new TcpMsg(BrokerResponseCode.BROKER_UPDATE_OFFSET_RESP.getCode(), JSON.toJSONBytes(consumeMsgAckRespDTO)));
            return;
        }
        List<ConsumerInstance> consumerInstances = consumeInstanceMap.get(consumerGroup);
        if (CollectionUtils.isEmpty(consumerInstances)) {
            event.getChannelHandlerContext().writeAndFlush(new TcpMsg(BrokerResponseCode.BROKER_UPDATE_OFFSET_RESP.getCode(), JSON.toJSONBytes(consumeMsgAckRespDTO)));
            return;
        }
        String currentConsumerReqId = consumeMsgAckReqDTO.getIp() + ":" + consumeMsgAckReqDTO.getPort();
        ConsumerInstance matchInstance = consumerInstances.stream().filter(item -> {
            return item.getConsumerRegId().equals(currentConsumerReqId);
        }).findAny().orElse(null);
        if (matchInstance == null) {
            event.getChannelHandlerContext().writeAndFlush(new TcpMsg(BrokerResponseCode.BROKER_UPDATE_OFFSET_RESP.getCode(), JSON.toJSONBytes(consumeMsgAckRespDTO)));
            return;
        }
        CommonCache.getConsumeQueueConsumeHandler().ack(topic, consumerGroup, queueId, ackCount);
        log.info("broker receive offset value, topic: {}, consumerGroup: {}, queueId: {}, ackCount: {}", topic, consumerGroup, queueId, ackCount);
        consumeMsgAckRespDTO.setAckStatus(AckStatus.SUCCESS.getCode());
        TcpMsg tcpMsg = new TcpMsg(BrokerResponseCode.BROKER_UPDATE_OFFSET_RESP.getCode(), JSON.toJSONBytes(consumeMsgAckRespDTO));
        event.getChannelHandlerContext().writeAndFlush(tcpMsg);
    }
}
